package javaVersion.learn.streamProcess;

import javaVersion.pojo.Pro5_Stream_User;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

/**
 * @author Wincoma
 * @date 2020/7/19 22:21
 * @description 实现MySQL数据源富函数
 */
public class Pro5_MySQLSource extends RichSourceFunction<Pro5_Stream_User> {
    Connection connection = null;
    PreparedStatement ps = null;


        @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection("jdbc:mysql//test", "root", "123456");
        String sql = "select id,name,age from t_user";
        ps = connection.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<Pro5_Stream_User> ctx) throws Exception {
        ResultSet rs = ps.executeQuery();
        while (rs.next()) {
            String id = rs.getString("id");
            String name = rs.getString("name");
            int age = rs.getInt("age");
            ctx.collect(new Pro5_Stream_User(id,name,age));
        }
    }

    //由于是有界数据源,当run方法中为false时,可以自动跳出循环,故下面的cancel不写亦可以.
    // 如果要写可以写 上面循环跳出的条件,当然,这个条件必须提前在声明在成员变量中.
    @Override
    public void cancel() {

    }

    @Override
    public void close() {
        System.out.println("关闭数据库连接:close!");
        try {
            ps.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        try {
            connection.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}
